package com.my.hbase;


import net.sf.json.JSONArray;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.log4j.Logger;

import java.io.IOException;

/**
 * hbase 基础操作类
 */
public class HBaseTemplate {

    private static  final Logger logger = Logger.getLogger(HBaseTemplate.class);

    private static Configuration configuration = null;

    private static Connection connection = null;

    private static HBaseAdmin hBaseAdmin = null;

    public  HBaseTemplate(){}

    static {
        init();
    }

    public static HBaseAdmin getAdmin(){
        return hBaseAdmin;
    }

    public static Configuration getConfiguration(){
        return configuration;
    }

    public  static  Connection getConnection(){
        return connection;
    }

    /**
     * 创建表
     * @param  tableName 表名称
     * @param  families 列族
     * @param  isForce 当表存在时是否强制创建表，true强制，false略过
     *                 注意：强制将清空原有数据
     */
    public static void createTable(String tableName, String[] families, boolean isForce) {

        try {
            if(!hBaseAdmin.tableExists(tableName)){//表不存在时的操作
                createTable(tableName, families);
            }else{ //表存在时的操作 isForce true强制，false略过
                if(isForce){
                    deleteTable(tableName);
                    createTable(tableName, families);
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    public static void deleteTable(String tableName){
        try {
            if(!hBaseAdmin.isTableDisabled(tableName)){
                hBaseAdmin.disableTable(tableName);
            }
            hBaseAdmin.deleteTable(tableName);
            logger.info("删除表["+tableName+"]成功！");
        } catch (IOException e) {
            logger.error("判断表失败！"+e.getMessage());
            e.printStackTrace();
        }
    }

    public static void save(){
        String tableName = "";
        String rowKey = "";
        try {
            Table table = connection.getTable(TableName.valueOf(tableName));

            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(""), Bytes.toBytes(""), Bytes.toBytes(""));

            table.put(put);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }



    public  static  void  main(String arg0[]) throws IOException {

       /*
       //创建表测试
       HBaseAdmin admin = HBaseTemplate.getAdmin();

        HTableDescriptor table = new HTableDescriptor(TableName.valueOf("table03"));

        HColumnDescriptor family = new HColumnDescriptor("family03");

        table.addFamily(family);

        admin.createTable(table);

        // HBaseTemplate.createTable("table03", new String[]{"family01","family02"}, true);
        */

       /* String[] a = new String[]{"s", "b"};

        System.out.println(JSONArray.fromObject(a).toString());*/

        //删除表测试
        HBaseTemplate.deleteTable("user");

    }

    /**
     * 初始化
     */
    private static void init() {
        configuration = HBaseConfiguration.create();
        configuration.setInt("hbase.rpc.timeout", Integer.valueOf(Config.getPropertie("hbase.rpc.timeout")));

        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            logger.error("HBase获取链接失败！"+e.getMessage());
            e.printStackTrace();
        }

        try {
            hBaseAdmin = (HBaseAdmin)connection.getAdmin();
        } catch (IOException e) {
            logger.error("获取HbaseAdmin失败！"+e.getMessage());
            e.printStackTrace();
        }
    }

    private static void createTable(String tableName, String[] families){
        HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));

        for(String family : families){
            HColumnDescriptor columnDescriptor = new HColumnDescriptor(family);
            tableDescriptor.addFamily(columnDescriptor);
        }

        try {
            hBaseAdmin.createTable(tableDescriptor);
            logger.info("创建表["+tableName+":"+ tableDescriptor.toString() +"成功!]");
        } catch (IOException e) {
            logger.error("创建表[" + tableName + ":" + JSONArray.fromObject(families).toString() + "失败!]");
            e.printStackTrace();
        }
    }



}
